package com.learn.kafka.transaction;

import com.learn.kafka.config.ProducerProperties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;

import java.util.UUID;

/**
 * 事务生产者（继承通用配置父类）
 * @author:         MI
 * @email:          448341911@qq.com
 * @createTime:     2025/8/9 12:21
 * @updateUser:     MI
 * @updateTime:     2025/8/9 12:21
 * @updateRemark:   修改内容
 * @version:        1.0
 */
public class TransactionProducer extends ProducerProperties {

    public static void main(String[] args) {
        new TransactionProducer().sendWithTransaction();
    }

    // 构造方法：初始化事务相关配置
    public TransactionProducer() {
        super(); // 调用父类构造方法，加载通用配置
        initTransactionConfig();
    }

    // 初始化事务特殊配置
    private void initTransactionConfig() {
        // 0 发出去就确认 | 1 leader 落盘就确认| all或-1 所有Follower同步完才确认(覆盖父类)
        setProperty(ProducerConfig.ACKS_CONFIG,"all");
        // 启用幂等性（事务前提）
        setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");

        // 设置唯一事务ID（使用UUID确保唯一性）
        setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, UUID.randomUUID().toString());
    }

    // 带事务的发送逻辑
    public void sendWithTransaction() {
        // 从父类获取生产者实例
        Producer<Integer, String> producer = super.createProducer();

        try {
            // 初始化事务
            producer.initTransactions();
            // 开始事务
            producer.beginTransaction();

            // 发送消息
            producer.send(new ProducerRecord<>("transaction-test", 1, "1"));
            producer.send(new ProducerRecord<>("transaction-test", 2, "2"));
            // 模拟异常：取消注释可测试事务回滚
            // int i = 1 / 0;
            producer.send(new ProducerRecord<>("transaction-test", 3, "3"));

            // 提交事务
            producer.commitTransaction();
            System.out.println("事务提交成功");
        } catch (KafkaException e) {
            // 事务回滚
            producer.abortTransaction();
            System.err.println("事务回滚：" + e.getMessage());
        } finally {
            // 关闭生产者
            producer.close();
        }
    }
}
